package cn.azzhu.day08

import cn.azzhu.day02.SensorSource
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api._
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment, tableConversions}
import org.apache.flink.table.functions.ScalarFunction
import org.apache.flink.table.runtime.operators.window.triggers.ElementTriggers.every
import org.apache.flink.types.Row
/**
 * Flink-Table-SQL：标量聚合和表聚合UDF函数
 * @author azzhu
 * @create 2020-09-23 22:49:35
 */
object ScalarFunctionExample {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val settings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    val tEnv = StreamTableEnvironment.create(env, settings)
    val stream = env.addSource(new SensorSource)

    val hashCode = new HashCode(10)

    //table写法
    val table = tEnv.fromDataStream(stream)

    table
      .select('id,hashCode('id))
      .toAppendStream[Row]
      .print()

    //sql写法
    tEnv.registerFunction("hashCode",hashCode)

    tEnv.createTemporaryView("sensor",table)
    tEnv
      .sqlQuery("select id,hashCode(id) from sensor")
      .toAppendStream[Row]
      .print()

    env.execute("ScalarFunctionExample")
  }

  class HashCode(factor:Int) extends ScalarFunction {
    def eval(s:String):Int = {
      s.hashCode * factor
    }
  }
}
